package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.BeeTableType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.google.common.collect.Lists;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DorisJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(StarRocksJdbcDialect.class);

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.DORIS;
    }

    public DorisJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public DorisJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    protected Driver getDriver() throws SQLException {
        return loadDriver(DataSourceType.MYSQL.getDriverClass());
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        if (schemaName.equalsIgnoreCase("_statistics_")
                || schemaName.equalsIgnoreCase("meta_model")
                || schemaName.equalsIgnoreCase("__internal_schema")) {
            return false;
        }
        return super.filterSchema(schemaName);
    }

    @Override
    protected String getDDLColumnName(BeeTableType type) {
        if (BeeTableType.TABLE == type) {
            return "Create Table";
        } else if (BeeTableType.VIEW == type) {
            return "Create View";
        } else {
            return "Create Materialized View";
        }
    }

    @Override
    public List<String> listSchemas(Connection connection) {
        // for MySQL, we need to list catalogs instead of schemas
        try (ResultSet resultSet = connection.getMetaData().getCatalogs()) {
            List<String> schemaNames = Lists.newArrayList();
            while (resultSet.next()) {
                String schemaName = resultSet.getString("TABLE_CAT");
                // skip internal schemas
                if (filterSchema(schemaName)) {
                    schemaNames.add(schemaName);
                }
            }
            return schemaNames;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        }
    }

    @Override
    protected String getSelectTablesSql(String tableName) {
        String sql = "select a.TABLE_NAME    AS TABLE_NAME,\n"
                + "       a.TABLE_SCHEMA  AS TABLE_SCHEMA,\n"
                + "       a.TABLE_COMMENT AS TABLE_COMMENT,\n"
                + "       a.TABLE_TYPE    AS TABLE_TYPE,\n"
                + "       a.TABLE_ROWS    AS TABLE_ROWS,\n"
                + "       a.DATA_LENGTH   AS DATA_SIZE,\n"
                + "       a.ENGINE        AS TABLE_ENGINE,\n"
                + "       null            AS TABLE_PARTITION,\n"
                + "       a.CREATE_TIME   AS CREATE_TIME,\n"
                + "       a.UPDATE_TIME   AS UPDATE_TIME\n"
                + "from information_schema.tables a\n"
                + "where a.TABLE_TYPE <> 'SYSTEM VIEW'\n"
                + "  and a.TABLE_SCHEMA = :schemaName";

        if (StringUtils.isNotBlank(tableName)) {
            sql += " AND TABLE_NAME = :tableName";
        }
        return sql;
    }

    @Override
    protected String getSelectColumnsSql() {
        return "select\n"
                + "    a.COLUMN_NAME,\n"
                + "    a.COLUMN_TYPE,\n"
                + "    a.DATA_TYPE,\n"
                + "    a.COLUMN_COMMENT,\n"
                + "    a.COLUMN_DEFAULT AS DEFAULT_VALUE,\n"
                + "    a.IS_NULLABLE,\n"
                + "    a.NUMERIC_PRECISION,\n"
                + "    a.NUMERIC_SCALE,\n"
                + "    a.CHARACTER_OCTET_LENGTH DATA_LENGTH,\n"
                + "    a.DATETIME_PRECISION\n"
                + "from INFORMATION_SCHEMA.COLUMNS a\n"
                + "where a.TABLE_SCHEMA = :schemaName and a.TABLE_NAME = :tableName\n"
                + "order by a.ORDINAL_POSITION ASC";
    }
}
